package id2210.system.peer.aggregation;

import java.util.ArrayList;
import java.util.Random;

import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.Timer;

import id2210.simulator.snapshot.Snapshot;
import id2210.system.common.PeerAddress;
import id2210.system.peer.cyclon.PartnersPort;
import id2210.system.peer.cyclon.PartnersRequest;
import id2210.system.peer.cyclon.PartnersResponse;

public final class Aggregation extends ComponentDefinition {

	Positive<PartnersPort> partnersPort = positive(PartnersPort.class);
	Positive<Network> networkPort = positive(Network.class);
	Positive<Timer> timerPort = positive(Timer.class);

	private PeerAddress self;
	private long period;
	private double num;
	private ArrayList<PeerAddress> cyclonPartners;
	private AggregationConfiguration aggregationConfiguration;
	Random rand;

	// -------------------------------------------------------------------
	public Aggregation() {
		cyclonPartners = new ArrayList<PeerAddress>();

		subscribe(handleInit, control);
		subscribe(handleRequestPartners, timerPort);
		subscribe(handleRecvPartners, partnersPort);
		subscribe(handleInfoExchangeRequest, networkPort);
		subscribe(handleInfoExchangeResponse, networkPort);
	}

	// -------------------------------------------------------------------
	Handler<AggregationInit> handleInit = new Handler<AggregationInit>() {
		public void handle(AggregationInit init) {
			rand = new Random();
			self = init.getSelf();
			num = init.getNum();
			aggregationConfiguration = init.getConfiguration();
			period = aggregationConfiguration.getPeriod();

			SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period,
					period);
			rst.setTimeoutEvent(new AggregationSchedule(rst));
			trigger(rst, timerPort);

			Snapshot.updateNum(self, num);
		}
	};

	// -------------------------------------------------------------------
	Handler<AggregationSchedule> handleRequestPartners = new Handler<AggregationSchedule>() {
		public void handle(AggregationSchedule event) {
			PartnersRequest request = new PartnersRequest();
			trigger(request, partnersPort);

			PeerAddress randomNeighbour = randomSelectANeighbour();
			if (randomNeighbour != null) {
				trigger(new InfoExchangeRequest(num, self, randomNeighbour),
						networkPort);
			}
		}
	};

	// -------------------------------------------------------------------
	Handler<PartnersResponse> handleRecvPartners = new Handler<PartnersResponse>() {
		public void handle(PartnersResponse event) {
			cyclonPartners = event.getPartners();
		}
	};

	Handler<InfoExchangeResponse> handleInfoExchangeResponse = new Handler<InfoExchangeResponse>() {
		public void handle(InfoExchangeResponse event) {
			update(event.getNum());
			Snapshot.updateNum(self, num);
		}
	};

	// -------------------------------------------------------------------
	Handler<InfoExchangeRequest> handleInfoExchangeRequest = new Handler<InfoExchangeRequest>() {
		public void handle(InfoExchangeRequest event) {
			trigger(new InfoExchangeResponse(num, self, event.getPeerSource()),
					networkPort);
			update(event.getNum());
			Snapshot.updateNum(self, num);
		}
	};

	private PeerAddress randomSelectANeighbour() {
		return this.cyclonPartners.isEmpty() ? null : this.cyclonPartners
				.get(rand.nextInt(this.cyclonPartners.size()));
	}

	private void update(double neighbourNum) {
		this.num = (this.num + neighbourNum) / 2;
	}
}
